package com.atbeijing.D06;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 双流connect 模拟开关流
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<SensorReading> sensorStream = env.addSource(new SensorSource());

        //Tuple2事件流
        DataStreamSource<Tuple2<String, Long>> switchStream = env.fromElements(
                //(感应器,开启时长)
                Tuple2.of("sensor_2", 10 * 1000L),
                Tuple2.of("sensor_1", 2 * 1000L)

        );

        sensorStream
                .keyBy(r -> r.id)
                //两个keyedStream进行connect,相同key的事件进入同一个下游算子,相当于join
                .connect(switchStream.keyBy(r -> r.f0))
                //每条流都有自己process方法,都有一个自己的SwitchFunction实例,里面的状态变量各不相关
                .process(new SwitchFunction())
                .print();

        env.execute();

    }

    public static class SwitchFunction extends CoProcessFunction<SensorReading, Tuple2<String, Long>, SensorReading> {

        //定义开关
        private ValueState<Boolean> sensorSwitch;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            sensorSwitch=getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("sensor2-switch",Types.BOOLEAN));
        }

        //处理sensorStream
        @Override
        public void processElement1(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
            //sensor_2的开关打开后,将数据下传
            if (sensorSwitch.value()!=null&&sensorSwitch.value()){
                out.collect(value);
            }
        }

        //处理switchStream
        @Override
        public void processElement2(Tuple2<String, Long> value, Context ctx, Collector<SensorReading> out) throws Exception {
            //开启sensor_2开关,并计时5s
            sensorSwitch.update(true);
            ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime()+value.f1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<SensorReading> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            // 关闭开关
            sensorSwitch.clear();
        }
    }
}
